package com.huan.maket;


import com.huan.bean.ChannelPromotionCount;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

//实现自定义全窗口函数
public class MarketingCountResult extends ProcessWindowFunction<Long, ChannelPromotionCount, Tuple, TimeWindow> {
    @Override
    public void process(Tuple tuple, Context context, Iterable<Long> elements, Collector<ChannelPromotionCount> out) throws Exception {
        String channel = tuple.getField( 0 ); //按照keyBy分组
        String behavior = tuple.getField( 1 );
        String windowEnd = new Timestamp( context.window().getEnd()).toString();
        Long count = elements.iterator().next();

        out.collect( new ChannelPromotionCount(channel,behavior,windowEnd,count) );
    }
}
